package com.atguigu.flink0922.chapter05.source;

import com.atguigu.flink0922.bean.WaterSensor;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.Socket;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/3/2 14:13
 */
public class Flink04_Source_Custom {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        env
            .addSource(new MySocketSource("hadoop162", 9999))
            .print();
        
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
    // 自定义一个从socket读取数据的source
    public static class MySocketSource implements SourceFunction<WaterSensor> {
        
        private final String host;
        private final int port;
        private boolean isCancel= false;
        private Socket socket;
        private BufferedReader reader;
    
        public MySocketSource(String host, int port) {
            this.host = host;
            this.port = port;
        }
        
        /*
        读取数据的代码, 应该写在这里面
        
        自动调用
         */
        @Override
        public void run(SourceContext<WaterSensor> ctx) throws Exception {
            System.out.println("run....");
            socket = new Socket(host, port);
            reader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
            String line = reader.readLine();
            while (line != null && !isCancel) {
                final String[] split = line.split(",");
                ctx.collect(new WaterSensor(split[0], Long.valueOf(split[1]), Integer.valueOf(split[2])));
                line = reader.readLine();
            }
            
        }
        
        /*
        将来在外部可以通过这个方法, 来取消数据的取
         */
        @Override
        public void cancel() {
            isCancel = true;
            
            if(reader != null) {
                try {
                    reader.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            if (socket != null) {
                try {
                    socket.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
    
            }
            
        }
    }
}
